package cn.com.dhcc.ddi.task.service;

import cn.com.dhcc.app.core.CmCommon;
import cn.com.dhcc.app.core.exception.DaoException;
import cn.com.dhcc.app.core.exception.ServiceException;
import cn.com.dhcc.app.core.handler.dbhandler.AbstractDao;
import cn.com.dhcc.app.core.handler.dbhandler.SingleTransationCircleWithOutResult;
import cn.com.dhcc.app.core.handler.modulehandler.WhereCondition;
import cn.com.dhcc.app.core.handler.pagerhandler.Pager;
import cn.com.dhcc.app.core.handler.pagerhandler.PagerData;
import cn.com.dhcc.app.core.handler.threadpoolhandler.ThreadPool;
import cn.com.dhcc.app.core.service.BaseService;
import cn.com.dhcc.app.core.service.Log;
import cn.com.dhcc.app.core.service.Log.LOGTYPE;
import cn.com.dhcc.app.core.tag.CommonStaticCode;
import cn.com.dhcc.app.core.util.CollectionUtil;
import cn.com.dhcc.app.core.util.JsonUtil;
import cn.com.dhcc.app.core.util.UUIDGenerator;
import cn.com.dhcc.app.pub.CM;
import cn.com.dhcc.app.pub.core.consts.AppIocBean;
import cn.com.dhcc.app.pub.core.consts.DdiProperties;
import cn.com.dhcc.app.pub.core.handler.FileAnalyzeHandler;
import cn.com.dhcc.app.pub.core.tag.staticcode.StaticCode;
import cn.com.dhcc.app.pub.core.tag.staticcode.StaticCode.FREQ_TYPE;
import cn.com.dhcc.app.pub.core.tag.staticcode.StaticCode.TASK_FREQ_TYPE;
import cn.com.dhcc.app.pub.core.tag.staticcode.StaticCode.TASK_STATUS;
import cn.com.dhcc.ddi.datamapping.dao.DataMappingDao;
import cn.com.dhcc.ddi.datamapping.vo.DataMapping;
import cn.com.dhcc.ddi.datasource.dao.DataSourceDao;
import cn.com.dhcc.ddi.datasource.vo.DataSourceConfVo;
import cn.com.dhcc.ddi.datasource.vo.DataSourceVo;
import cn.com.dhcc.ddi.datasource.vo.TableInfoVo;
import cn.com.dhcc.ddi.task.dao.*;
import cn.com.dhcc.ddi.task.service.extract.EDataSourceInterface;
import cn.com.dhcc.ddi.task.service.load.LDataSourceInterface;
import cn.com.dhcc.ddi.task.service.notify.NotifyInterface;
import cn.com.dhcc.ddi.task.service.transform.TDataInterface;
import cn.com.dhcc.ddi.task.vo.*;
import cn.com.dhcc.ddi.task.vo.TaskOntime.EXECUTE_TYPE;

import org.apache.log4j.Logger;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import org.springframework.util.StringUtils;

import java.sql.Timestamp;
import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.util.*;

/**
 * etl任务Service
 * @日期：2016/11/10 17:21
 * @作者：张洪涛
 * @版权所有：东华软件
 * @版本：1.0
 */
@Service
public class TaskService extends BaseService {
	private static Logger logger = Log.getLogger(LOGTYPE.DDI);
	private static final Object sendLock = new Object();
	@Autowired
	private TaskDao taskDao;
	@Autowired
	private TaskOntimeDao ontimeDao;
	@Autowired
	private TaskExecuteLogDao taskExecuteLogDao;
	@Autowired
	private TaskPeriodDao periodDao;
	@Autowired
	private TaskActiveLogDao taskActiveLogDao;
	@Autowired
	private DataMappingDao dmDao;
	@Autowired
	private DataSourceDao dsDao;
	@Autowired
	private TableLogDao tableLogDao;
	@Autowired
	private TaskTabDao taskTabDao;

	/**
	 *
	 * @方法: executeTasks
	 * @描述: 运行待执行的任务
	 * @throws ServiceException
	 * @throws
	 */
	public void executeTasks() throws ServiceException{
		//获取待运行任务.
		synchronized (sendLock) {
			List<Task> tasks = getNeedExecuteTask();
			if(CollectionUtil.hasElement(tasks)){
				for (final Task task : tasks) {
					ThreadPool.execute(FileAnalyzeHandler.THREAD_POOL_STRUC, new Runnable() {
						@Override
						public void run() {
						try {
							logger.info("开始执行etl任务：" + task.getTaskName());
							executeTask(task);
						} catch (ServiceException e) {
							logger.error("执行etl任务失败", e);
						} catch (ParseException e) {
							logger.error("执行etl任务异常", e);
						}
						}
					});
				}
			}
		}
	}

	/**
	 *
	 * @方法: executeTask
	 * @描述: 执行任务
	 * @param task
	 * @throws ServiceException
	 * @throws ParseException
	 * @throws
	 */
	private void executeTask(Task task) throws ServiceException, ParseException{
		//一次性任务
		if(task.getTaskFreqType().equalsIgnoreCase(TASK_FREQ_TYPE.ONETIME.getCode())){
			if(!onTimeTaskNeedSend(task.getId())){
				return;
			}
		}else{//周期性任务
			logger.info("获取需要执行的周期性任务：" + task.getTaskName());
			if(!proidTaskNeedSend(task)){
				return;
			}
		}
		//记日志
		//有子任务要发送就把主任务和子任务的状态设置为发送中，并记录日志
		Date now = new Date();
		task.setStatus(TASK_STATUS.ING.getCode());
		//主任务日志
		TaskExecuteLog taskExecuteLog =  new TaskExecuteLog();
		String id = UUIDGenerator.getUUID();
		taskExecuteLog.setId(id);
		taskExecuteLog.setMtime(now);
		taskExecuteLog.setLinkStartTime(now);
		taskExecuteLog.setTaskId(task.getId());
		try {
			int iInsertNum = taskExecuteLogDao.insert(taskExecuteLog);
			logger.info("插入表t_birs_struc_task_log更新记录数为：" + iInsertNum);
			int iUpdateNum = taskDao.update(task);
			logger.info("更新t_birs_struc_stask记录数为：" + iUpdateNum + ";更新状态为: " + task.getStatus());
		} catch (DaoException e) {
			throw new ServiceException(task.getTaskName()+"更新任务运行状态日志出错",e);
		}


		//执行任务
		try {
			TaskActiveLog log  = taskActiveLogDao.getByTaskId(task.getId());
			if(log == null){//如果第一次运行，则新增任务运行状态日志
				log = new TaskActiveLog();
				log.setId(UUIDGenerator.getUUID());
				log.setTaskId(task.getId());
			}
			log.setMtime(new Date());
			taskActiveLogDao.insertOrUpdate(log);
			//通过数据映射获取对应源端数据源
			DataMapping dm = dmDao.get(task.getMpId());
			DataSourceVo ds = dsDao.get(dm.getSrcDsId());

			NotifyInterface notify =  getNotify(ds);
            assert notify != null;
			//1.抽取数据
			EDataSourceInterface extractImpl = getExtractDataHandler(ds);
            assert extractImpl != null;
            Result extractResult = extractImpl.extractData(task,taskExecuteLog.getId());
			if (extractResult.getStatus().equalsIgnoreCase(CommonStaticCode.SUCCESSFAIL.FAIL.getStatus())){
				notify.reciveNotify(taskExecuteLog.getId(),extractResult.getStatus(),extractResult.getMsg());
				return;
			}
			//2.映射数据
			TDataInterface transformImpl = getTransformDataHandler(dm.getMappingType());
            assert transformImpl != null;
			Result transformResult = transformImpl.transformData(task,taskExecuteLog.getId());
			if (transformResult.getStatus().equalsIgnoreCase(CommonStaticCode.SUCCESSFAIL.FAIL.getStatus())){
				notify.reciveNotify(taskExecuteLog.getId(),transformResult.getStatus(),transformResult.getMsg());
				return;
			}
			//3.加载数据
			LDataSourceInterface loadImpl = getLoadDataHandler(ds);
            assert loadImpl != null;
			Result loadResult = loadImpl.loadData(task,taskExecuteLog.getId());
			if (loadResult.getStatus().equalsIgnoreCase(CommonStaticCode.SUCCESSFAIL.FAIL.getStatus())){
                notify.reciveNotify(taskExecuteLog.getId(),loadResult.getStatus(),loadResult.getMsg());
				return;
			}

			//4.交换成功后，修改任务状态

			notify.reciveNotify(taskExecuteLog.getId(), CommonStaticCode.SUCCESSFAIL.SUCCESS.getStatus(),"");
		} catch (DaoException e) {
			logger.error("etl任务执行异常",e);
			throw new ServiceException(task.getTaskName()+"更新任务运行状态日志出错",e);
		}
	}



	/**
	 * 获取待执行任务
	 * @return
	 * @throws ServiceException
     */
	private List<Task> getNeedExecuteTask() throws ServiceException{
		WhereCondition condition = new WhereCondition();
		//大任务发送中，但是里面有的小任务执行完了也可以执行，不用等上一批执行完毕
		condition.where().in("status", new String[]{TASK_STATUS.WAIT.getCode(),TASK_STATUS.ERROR.getCode()});
		try {
			return CM.getDao().listModule("获取待执行任务", Task.class, condition);
		} catch (DaoException e) {
			throw new ServiceException(e);
		}
	}

    /**
     * 判断一次性任务是否需要运行
     * @param taskId
     * @return
     * @throws ServiceException
     */
	private boolean onTimeTaskNeedSend(String taskId) throws ServiceException {
		TaskOntime ontime;
		try {
			ontime = ontimeDao.getTaskOntime(taskId);
		} catch (DaoException e) {
			throw new ServiceException(e);
		}
		if(ontime.getExecuteType().equals(EXECUTE_TYPE.IMME.getCode())){
			return true;
		}else if(ontime.getExecuteType().equals(EXECUTE_TYPE.TIMING.getCode())){
			Date dateTask = ontime.getExecuteTime();
			Date dateNow = new Date();
			if (dateTask.compareTo(dateNow) <= 0) {
				return true;
			}
		}
		return false;
	}


    /**
     * 判断周期性任务是否要执行
     * @param task
     * @return
     * @throws ServiceException
     * @throws ParseException
     */
	private boolean proidTaskNeedSend(Task task) throws ServiceException, ParseException {
		logger.info("进入方法proidTaskNeedSend().");
		TaskExecuteLog taskExecuteLog;
		try {
			taskExecuteLog = taskExecuteLogDao.getLastExecuteLog(task.getId());
		} catch (DaoException e) {
			logger.error("获取最后一次执行的日志信息错误：" + e.getMessage());
			throw new ServiceException(e);
		}
		TaskPeriod taskPeriod;
		try {
			taskPeriod = periodDao.getTaskPeriod(task.getId());

			Date cutOffTime = taskPeriod.getCutOffTime();//截至时间
			Calendar now= Calendar.getInstance();
			Calendar cutOffTimeCal = Calendar.getInstance();
			if(cutOffTime!=null){
				cutOffTimeCal.setTime(cutOffTime);
			}
			if(cutOffTime!=null&&now.after(cutOffTimeCal)){//过了截至时间
				if(task.getStatus().equalsIgnoreCase(TASK_STATUS.ERROR.getCode())){//虽然过期，但是失败的还是要发送
					return true;
				}
			}else{//未过截至时间，计算是否要执行
				//指定频率 1 间隔 2每天3 每周4 每月
				//1 间隔 和上次执行时间比较
				Date lastStartTime;
				if(taskExecuteLog == null){
					String dateString = "1970-01-01 00:00:00";//上次运行时间
					SimpleDateFormat sdf =   new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
					lastStartTime = sdf.parse(dateString);
				}else{
					lastStartTime = taskExecuteLog.getLinkStartTime();
				}

				Calendar cLastStart = Calendar.getInstance();
				if(lastStartTime!=null ){
					cLastStart.setTime(lastStartTime);
				}
				//间隔
				if(taskPeriod.getFreq().equals(FREQ_TYPE.INTERVAL.getCode())){
					if(taskExecuteLog == null ){
						logger.info("开始执行。。" + taskPeriod.getTaskId());
						return true;
					}
					int intervalSecond = Integer.parseInt(taskPeriod.getIntervalVallue());
					cLastStart.add(Calendar.SECOND, intervalSecond);
					if((cLastStart.compareTo(now) <= 0)){
						logger.info("开始执行。。" + taskPeriod.getTaskId());
						return true;
					}
				}
				String executeTime = taskPeriod.getExecuteTime();
				//每天
				if(taskPeriod.getFreq().equals(FREQ_TYPE.DAYLY.getCode())){
					String[] HMS = executeTime.split(":");

					if(HMS.length >= 3){
						Calendar cThis= Calendar.getInstance();
						cThis.setTime(now.getTime());
						cThis.set(Calendar.HOUR_OF_DAY, Integer.parseInt(HMS[0]));
						cThis.set(Calendar.MINUTE, Integer.parseInt(HMS[1]));
						cThis.set(Calendar.SECOND, Integer.parseInt(HMS[2]));
						now.set(Calendar.SECOND, Integer.parseInt(HMS[2]));
						if(taskExecuteLog == null ){
							return cThis.compareTo(now) == 0;
						}
						cLastStart.add(Calendar.HOUR_OF_DAY, 23);
						cLastStart.add(Calendar.MINUTE, 59);
						if((cThis.compareTo(cLastStart) > 0) && (cThis.compareTo(now) == 0)){
							return true;
						}
					}
				}
				//每周
				if(taskPeriod.getFreq().equals(FREQ_TYPE.WEEKLY.getCode())){
					String[] HMS = executeTime.split(":");
					Integer dayOfWeek = Integer.parseInt(taskPeriod.getIntervalVallue()) % 7;
					if((HMS.length >= 3) && (now.get(Calendar.DAY_OF_WEEK) == dayOfWeek)){
						Calendar cThis= Calendar.getInstance();
						cThis.setTime(now.getTime());
						cThis.set(Calendar.HOUR_OF_DAY, Integer.parseInt(HMS[0]));
						cThis.set(Calendar.MINUTE, Integer.parseInt(HMS[1]));
						cThis.set(Calendar.SECOND, Integer.parseInt(HMS[2]));
						now.set(Calendar.SECOND, Integer.parseInt(HMS[2]));
						if(taskExecuteLog == null ){
							return cThis.compareTo(now) == 0;
						}
						cLastStart.add(Calendar.DATE, 6);
						cLastStart.add(Calendar.HOUR_OF_DAY, 23);
						cLastStart.add(Calendar.MINUTE, 59);
						if((cThis.compareTo(cLastStart) > 0) && (cThis.compareTo(now) == 0) ){
							return true;
						}
					}
				}
				//每周
				if(taskPeriod.getFreq().equals(FREQ_TYPE.MONTHLY.getCode())){
					String[] HMS = executeTime.split(":");
					Integer dayOfMonth = Integer.parseInt(taskPeriod.getIntervalVallue());
					//如果本月没有指定日期，则在本月最后一天执行
					if(now.getActualMaximum(Calendar.DAY_OF_MONTH) > dayOfMonth){
						dayOfMonth = now.getActualMaximum(Calendar.DAY_OF_MONTH);
					}
					if((HMS.length >= 3) && ((now.get(Calendar.DAY_OF_MONTH) == dayOfMonth))){
						Calendar cThis= Calendar.getInstance();
						cThis.setTime(now.getTime());
						cThis.set(Calendar.HOUR_OF_DAY, Integer.parseInt(HMS[0]));
						cThis.set(Calendar.MINUTE, Integer.parseInt(HMS[1]));
						cThis.set(Calendar.SECOND, Integer.parseInt(HMS[2]));
						now.set(Calendar.SECOND, Integer.parseInt(HMS[2]));
						if(taskExecuteLog == null ){
							return cThis.compareTo(now) == 0;
						}
						cLastStart.add(Calendar.DATE, 27);
						cLastStart.add(Calendar.HOUR_OF_DAY, 23);
						cLastStart.add(Calendar.MINUTE, 59);
						if((cThis.compareTo(cLastStart) > 0) && (cThis.compareTo(now) == 0)){
							return true;
						}
					}
				}
			}
		} catch (DaoException e) {
			throw new ServiceException(e);
		}
		logger.info("执行结束proidTaskNeedSend().");
		return false;
	}

	/**
	 * 获取数据源抽取实现类
	 * @param dbType
	 * @return
     */
	private EDataSourceInterface getExtractDataHandler(DataSourceVo ds) {
		//数据源为数据库
		if(StaticCode.DS_TYPE.DATA_BASE.getCode().equalsIgnoreCase(ds.getType())){
			DataSourceConfVo confVo = JsonUtil.toObject(ds.getConf(),DataSourceConfVo.class);
			String dbType = confVo.getDbType();
			if(StaticCode.DB_TYPE.ORACLE.getCode().equalsIgnoreCase(dbType) ){
				return (EDataSourceInterface) CM.getBeanByName(AppIocBean.E_DATASOURCE_ORACLE_IMPL);
			} else if (StaticCode.DB_TYPE.MYSQL.getCode().equalsIgnoreCase(dbType)){
				return (EDataSourceInterface) CM.getBeanByName(AppIocBean.E_DATASOURCE_MYSQL_IMPL);
			} else if (StaticCode.DB_TYPE.SQLSERVER.getCode().equalsIgnoreCase(dbType) || StaticCode.DB_TYPE.SQLSERVEREXT.getCode().equalsIgnoreCase(dbType)){
				return (EDataSourceInterface) CM.getBeanByName(AppIocBean.E_DATASOURCE_SQLSERVER_IMPL);
			}
		}if(StaticCode.DS_TYPE.LOCAL_DISK.getCode().equalsIgnoreCase(ds.getType())){
			return (EDataSourceInterface) CM.getBeanByName(AppIocBean.E_DATASOURCE_LOCALDISK_IMPL);
		}

		return null;
	}

	/**
	 * 获取数据源抽取实现类
	 * @param dbype
	 * @return
	 */
	private LDataSourceInterface getLoadDataHandler(DataSourceVo ds) {
		//数据源为数据库
		if(StaticCode.DS_TYPE.DATA_BASE.getCode().equalsIgnoreCase(ds.getType())){
			DataSourceConfVo confVo = JsonUtil.toObject(ds.getConf(),DataSourceConfVo.class);
			String dbType = confVo.getDbType();
			if(StaticCode.DB_TYPE.ORACLE.getCode().equalsIgnoreCase(dbType) ){
				return (LDataSourceInterface) CM.getBeanByName(AppIocBean.L_DATASOURCE_ORACLE_IMPL);
			} else if (StaticCode.DB_TYPE.MYSQL.getCode().equalsIgnoreCase(dbType)){
				return (LDataSourceInterface) CM.getBeanByName(AppIocBean.L_DATASOURCE_MYSQL_IMPL);
			} else if (StaticCode.DB_TYPE.SQLSERVER.getCode().equalsIgnoreCase(dbType) || StaticCode.DB_TYPE.SQLSERVEREXT.getCode().equalsIgnoreCase(dbType)){
				return (LDataSourceInterface) CM.getBeanByName(AppIocBean.L_DATASOURCE_SQLSERVER_IMPL);
			}
		}else if(StaticCode.DS_TYPE.LOCAL_DISK.getCode().equalsIgnoreCase(ds.getType())){
			return (LDataSourceInterface) CM.getBeanByName(AppIocBean.L_DATASOURCE_LOCALDISK_IMPL);
		}

		return null;
	}

	/**
	 * 获取通知实现类
	 * @param ds
	 * @return
     */
	private NotifyInterface getNotify(DataSourceVo ds) {
		//数据源为数据库
		if(StaticCode.DS_TYPE.DATA_BASE.getCode().equalsIgnoreCase(ds.getType())){
			DataSourceConfVo confVo = JsonUtil.toObject(ds.getConf(),DataSourceConfVo.class);
			String dbType = confVo.getDbType();
			if(StaticCode.DB_TYPE.ORACLE.getCode().equalsIgnoreCase(dbType) ){
				return (NotifyInterface) CM.getBeanByName(AppIocBean.NOTIFY_ORACLE_IMPL);
			} else if (StaticCode.DB_TYPE.MYSQL.getCode().equalsIgnoreCase(dbType)){
				return (NotifyInterface) CM.getBeanByName(AppIocBean.NOTIFY_MYSQL_IMPL);
			} else if (StaticCode.DB_TYPE.SQLSERVER.getCode().equalsIgnoreCase(dbType)||StaticCode.DB_TYPE.SQLSERVEREXT.getCode().equalsIgnoreCase(dbType)){
				return (NotifyInterface) CM.getBeanByName(AppIocBean.NOTIFY_SQLSERVER_IMPL);
			}
		}else if(StaticCode.DS_TYPE.LOCAL_DISK.getCode().equalsIgnoreCase(ds.getType())){
            return (NotifyInterface) CM.getBeanByName(AppIocBean.NOTIFY_LOCALDISK_IMPL);
        }

		return null;
	}
	/**
	 * 获取数据源抽取实现类
	 * @param dbType
	 * @return
	 */
	private TDataInterface getTransformDataHandler(String code) {
		//数据源为数据库
		return CM.getBean(DataMapping.MP_TYPE.find(code).getHandler());
	}
	
	/**
	 * 获取周期性任务list
	 * @param  searchData where查询条件
	 * @return 所有状态不是“已完成”的周期性任务
	 * @throws ServiceException
	 */
	public PagerData<Task> getPeriodPagerList(Pager pager, WhereCondition searchData) throws ServiceException{
		try {
			return taskDao.getPeriodPagerList(pager, searchData);
		} catch (DaoException e) {
			throw new ServiceException("获取周期性任务失败", e);
		}
	}
	
	public List<Task> getPeriodList(WhereCondition searchParameter) throws ServiceException {
		try {
			return taskDao.getPeriodList(searchParameter);
		} catch (DaoException e) {
			throw new ServiceException("获取周期性任务失败", e);
		}
	}
	
	/**
	 * 获取一次性任务list
	 * @param  searchData where查询条件
	 * @return 所有状态不是“已完成”的周期性任务
	 * @throws ServiceException
	 */
	public PagerData<Task> getOnTimePagerList(Pager pager, WhereCondition searchData) throws ServiceException{
		try {
			return taskDao.getOnTimePagerList(pager, searchData);
		} catch (DaoException e) {
			throw new ServiceException("获取周期性任务失败", e);
		}
	}
	
	/**
	 * 获取历史任务列表
	 * @param pager      分页类
	 * @param searchData 条件sql
	 * @return 获取历史任务列表
	 * @throws ServiceException
	 */
	public PagerData<Task> getHistoryList(Pager pager, WhereCondition searchData) throws ServiceException {
		try {
			return taskDao.getHistoryList(pager, searchData);
		} catch (DaoException e) {
			throw new ServiceException("获取历史任务列表失败", e);
		}
	}
	
	public List<Task> getTaskList(WhereCondition condition) throws ServiceException {
		try {
			return taskDao.list(condition);
		} catch (DaoException e) {
			throw new ServiceException("获取历史任务列表失败", e);
		}
	}
	
	public void insertOrUpdateTask(Task task) throws ServiceException {
		try {
			CM.getDao().saveOrUpdateModuleByIdExecuteNull("添加或者修改任务", task);
		} catch (DaoException e) {
			throw new ServiceException(e);
		}
	}
	
	/**
	 * 验证属性唯一性
	 * @param property 属性名称
	 * @param newValue 新值
	 * @param oldValue 旧值
	 * @return true唯一 false不唯一
	 * @throws ServiceException
	 */
	public boolean isPropertyUnique(String property, String newValue, String oldValue)
			throws ServiceException {
		try {
			return taskDao.isPropertyUnique(property, newValue, oldValue);
		} catch (DaoException e) {
			throw new ServiceException("验证属性唯一性失败", e);
		}
	}
	
	/**
	 * 根据id获取任务
	 * @param  id 任务id
	 * @return 任务
	 * @throws ServiceException 
	 */
	public Task get(String id) throws ServiceException {
		try {
			return taskDao.get(id);
		} catch (DaoException e) {
			throw new ServiceException("根据id获取任务失败", e);
		}
	}
	
	/**
	 * 更新任务
	 * @param  vo 需要更新任务类
	 * @return 更新成功数
	 * @throws ServiceException
	 */
	public int update(Task vo) throws ServiceException {
		try {
			return taskDao.update(vo);
		} catch (DaoException e) {
			throw new ServiceException("更新任务失败", e);
		}
	}
	
	public int insertOrUpdate(Task vo) throws ServiceException {
		try {
			return taskDao.insertOrUpdate(vo);
		} catch (DaoException e) {
			throw new ServiceException("更新任务失败", e);
		}
	}
	
	/**
	 * 删除任务
	 * @param vo
	 * @return
	 * @throws ServiceException
	 */
	public int delete(Task vo) throws ServiceException {
		try {
			return taskDao.delete(vo);
		} catch (DaoException e) {
			throw new ServiceException("删除任务失败", e);
		}
	}
	
	/**
	 * 根据任务id删除任务
	 * @param id
	 * @return
	 * @throws ServiceException 
	 */
	public int deleteById(String id) throws ServiceException {
		try {
			return taskDao.deleteById(id);
		} catch (DaoException e) {
			throw new ServiceException("删除任务失败", e);
		}
	}
	
	/**
	 * 批量删除任务
	 * @param ids 任务id列表
	 * @return 成功个数
	 * @throws ServiceException
	 */
	public int batchDeleteByid(String[] ids) throws ServiceException {
		try {
			return taskDao.batchDeleteById(ids);
		} catch (DaoException e) {
			throw new ServiceException(e.getMessage(), e);
		}
	}
	
	/**
	 * 彻底删除历史任务
	 * @param id
	 * @throws ServiceException 
	 */
	public void deleteHistory(final String id) throws ServiceException {
			
			final AbstractDao dao = CmCommon.getDao();
			try {
				dao.doInSingleTransationCircle("添加权限", new SingleTransationCircleWithOutResult() {
					@Override
					public void actionInCircle() throws RuntimeException {
						try {
							//删除主任务
							taskDao.deleteById(id);
							
							//删除对应类型任务
							ontimeDao.deleteByTaskId(id);
							periodDao.deleteByTaskId(id);
							
							//删除任务关联表
							taskTabDao.deleteByTaskId(id);
							
							//删除任务日志
							taskExecuteLogDao.deleteByTaskId(id);
							
							//删除任务表日志
							tableLogDao.deleteByTaskId(id);
							
							//删除任务活动日志表
							taskActiveLogDao.deleteByTaskId(id);
							
							//删除统计表中对应任务
							//TODO: 删除统计
							//task
							
						} catch (DaoException e) {
							throw new RuntimeException(e);
						}
					}
				});
			} catch (DaoException e) {
				throw new ServiceException(e);
			}	
	}
	
	public void startTimeTasks() throws ServiceException {
		try {
			List<Task> ontimeList = taskDao.getTimedOntimeTaskList();
			List<Task> periodList = taskDao.getTimedPeriodTaskList();
			
			for(Task task: ontimeList) {
				Task vo = this.get(task.getId());
				vo.setStatus("2");
				this.update(vo);
			}
			
			for(Task task: periodList) {
				Task vo = this.get(task.getId());
				vo.setStatus("2");
				this.update(vo);
			}
			
		} catch (DaoException e) {
			throw new ServiceException(e);
		}	
	}


	/**
	 * 检测发送任务是否已中止
	 * @throws DaoException
	 */
	public void testingTaskStatus() throws ServiceException{
		try {
			//当出现任务状态为执行中时，如果超过n小时未更新活动日志表mtime字段值，则对任务状态修正
			WhereCondition condition = new WhereCondition();
			condition.where().in("status", new String[]{TASK_STATUS.ING.getCode()});
			List<Task> sendingTasks = taskDao.getTasks(condition);

			for (Iterator<Task> iterator = sendingTasks.iterator(); iterator.hasNext();) {
				Task task = iterator.next();
				TaskActiveLog log = taskActiveLogDao.getByTaskId(task.getId());
				if(log ==null){//老任务，没有任务活动日志，则新一条日志
					TaskExecuteLog taskExecuteLog = taskExecuteLogDao.getLastDataExecuteLog(task.getId());
					log = new TaskActiveLog();
					log.setId(UUIDGenerator.getUUID());
					log.setTaskId(task.getId());
					//上一次有数据的发送任务发送时间做为最新活时间，如果一次没运行成功，则将任务修改时间做为上次活动时间
					if(taskExecuteLog != null){
						log.setMtime(taskExecuteLog.getLinkStartTime());
					}else{
						log.setMtime(task.getMtime());
					}
					taskActiveLogDao.insertOrUpdate(log);
				}
				Calendar lastTime = Calendar.getInstance();
				Calendar now= Calendar.getInstance();
				if(log.getMtime()!=null ){
					lastTime.setTime(log.getMtime());
				}
				//如果超过 CHECK_STATUS_TIME 个小时时间字段未更新，则认为任务已经终止，需要任务状态重置，强制上次任务失败
				lastTime.add(Calendar.HOUR, DdiProperties.APP_CHECKSTATUSTIME);
				if((lastTime.compareTo(now) <= 0)){
					resetTask(task);
				}

			}


			//当出现任务状态为待执行时，而表发送日志temp_lasttime有值时，则将temp_lasttime值清空,子任务状态判断是否为待执行和执行异常，如果不是则改为待执行和执行异常
			condition = new WhereCondition();
			condition.where().in("status", new String[]{TASK_STATUS.WAIT.getCode(),TASK_STATUS.ERROR.getCode()});
			List<Task> waitTasks = taskDao.getTasks(condition);

			for (Iterator<Task> iterator = waitTasks.iterator(); iterator.hasNext();) {
				Task task = iterator.next();
				List<TaskExecuteLog> plogs = taskExecuteLogDao.getSendingTaskExecuteLog(task.getId());
				if(plogs !=null && plogs.size()>0){
					TaskExecuteLog plog = plogs.get(0);
					if(plog.getLogStatus() == null || "".equals(plog.getLogStatus())){
						if(task.getStatus().equals(TASK_STATUS.WAIT.getCode())){
							plog.setLogStatus(CommonStaticCode.SUCCESSFAIL.SUCCESS.getStatus());
						}else{
							plog.setLogStatus(CommonStaticCode.SUCCESSFAIL.FAIL.getStatus());
						}
						taskExecuteLogDao.update(plog);
					}
				}
				//保证表发送日志temp_lasttime为空
				List<TableLog> sTableLogs  = tableLogDao.listByTaskId(task.getId());
				for (Iterator<TableLog> iteratorTab = sTableLogs.iterator(); iteratorTab.hasNext();) {
					TableLog tableLog = iteratorTab.next();
					if(tableLog.getTempLastValue() != null){
						tableLog.setTempLastValue(null);
						tableLogDao.update(tableLog);
					}
				}
			}
		} catch (DaoException e) {
			throw new ServiceException("修改任务状态失败", e);
		}
	}

	/**
	 * 修改任务状态，强制上次执行为失败
	 * @param task
	 * @throws ServiceException
	 */
	public void resetTask(Task task) throws ServiceException{
		try {
			//修改大任务日志状态
			List<TaskExecuteLog> plogs = taskExecuteLogDao.getSendingTaskExecuteLog(task.getId());
			if(plogs !=null && plogs.size()>0){
				TaskExecuteLog plog = plogs.get(0);
				plog.setLogStatus(CommonStaticCode.SUCCESSFAIL.FAIL.getStatus());
				plog.setLinkEndTime(new Date());
				taskExecuteLogDao.update(plog);
			}


			List<TableLog> tableLogs = tableLogDao.listByTaskId(task.getId());
			for (Iterator<TableLog> iteratorTab = tableLogs.iterator(); iteratorTab.hasNext();) {
				TableLog sTableLog = iteratorTab.next();
				sTableLog.setTempLastValue(null);
				tableLogDao.update(sTableLog);
			}
			task.setStatus(TASK_STATUS.ERROR.getCode());
			taskDao.update(task);
		} catch (DaoException e) {
			throw new ServiceException("修改任务状态失败", e);
		}
	}
	
	public List<Task> getTasks(WhereCondition condition) throws ServiceException{
		try {
			return taskDao.getTasks(condition);
		} catch (DaoException e) {
			throw new ServiceException("获取etl任务", e);
		}
	}

	public void resetTasks() throws ServiceException{
		try {
			WhereCondition condition = new WhereCondition();
			condition.where().in("status", new String[]{TASK_STATUS.ING.getCode(),TASK_STATUS.ERROR.getCode()});
			List<Task> tasks = taskDao.getTasks(condition);
			for (Iterator<Task> iterator = tasks.iterator(); iterator.hasNext();) {
				Task task = iterator.next();
				resetTask(task);
			}
		} catch (DaoException e) {
			throw new ServiceException("修改任务状态失败", e);
		}
	}
}
